﻿using Sunxsh.Es.Common.Util;
using Newtonsoft.Json;
using RabbitMQ.Client;

namespace Sunxsh.Es.Common.Rabbitmq;

/// <summary>
/// 简单的消息队列封装, 使用默认switch, 生产者消费者模型
/// </summary>
public class MQLiteClient<T>
{
	MQConn _conn;
	string _queue;

	IModel _channel;

	Task _initTask;
	Task _waitTask;

	internal MQLiteClient(MQConn conn, string queue)
	{
		_conn  = conn;
		_queue = queue;

		_initTask = Task.Run(connect);                          //连接服务器的任务， 会一直重试
		_waitTask = Task.WhenAny(_initTask, Task.Delay(15_000)); //发送时候等待连接，最多等待15秒
	}


	public MQLiteClient(string queue)
		: this(MQConn.Huasky, queue)
	{
	}


	async Task connect()
	{
		var channel = await _conn.CreateChannel();
		lock (this)
		{
			channel.QueueDeclare(_queue);
			_channel = channel;
		}
	}

	/// <summary>
	/// 给队列发送消息
	/// </summary>
	public void Publish(T data)
	{
		if (!_waitTask.IsCompleted)
		{
			_waitTask.Wait();
		}
		
		if (_channel == null)
		{
			LoggerHelper.Error("MQ 未连接 [{0}]", _queue);
			throw new InvalidOperationException("MQ 未连接");
		}

		lock (this) //线程安全
		{
			_channel.Publish(string.Empty, _queue, data);
		}
	}

	/// <summary>
	/// 从队列获取消息， 多个进程注册同一个队列为多消费者模型
	/// 同一个进程多次注册同一个队列的回调会报错
	/// </summary>
	public async void Subscribe(ushort fetchSize, Func<T, Task> onData)
	{
		await _initTask;
		lock (this)
		{
			_channel.Subscribe(_queue, fetchSize, para =>
			{
				var data = JsonConvert.DeserializeObject<T>(para.Content);
				return onData(data);
			});
		}
	}


	/// <summary>
	/// 批量处理消息，消息处理完或到达最大次数后返回，不能和Subscribe同时使用
	/// </summary>
	public async Task BatchConsume(Func<T, Task> onData, int? maxData = null)
	{
		if (!_waitTask.IsCompleted)
		{
			_waitTask.Wait();
		}

		if (_channel == null)
		{
			//指定的时间内没有连接上服务器
			return;
		}

		int count = 0;
		while (true)
		{
			var isSuccess = await _channel.BasicConsume(_queue, onData);
			if (!isSuccess)
			{
				return;
			}

			if (maxData != null && maxData < ++count)
			{
				return;
			}
		}
	}
}